package com.yuan;

import java.io.IOException;
import java.io.PrintStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

/**
 * 自定义路径过滤器，用于过滤掉文件名满足特定条件的文件，并跳过目录。
 */
class MyPathFilter implements PathFilter {
    String reg = null; // 正则表达式，用于匹配文件名

    /**
     * 构造函数，初始化正则表达式。
     * @param reg 用于匹配文件名的正则表达式
     */
    MyPathFilter(String reg) {
        this.reg = reg;
    }

    /**
     * 判断给定路径是否符合过滤条件。
     * @param path 要检查的路径
     * @return 如果路径不符合正则表达式且不是目录，则返回 true；否则返回 false
     */
    public boolean accept(Path path) {
        try {
            FileSystem fs = path.getFileSystem(new Configuration());
            FileStatus status = fs.getFileStatus(path);

            // 如果是目录，则跳过
            if (status.isDirectory()) {
                return false;
            }

            // 如果路径的字符串形式不匹配正则表达式，则返回 true（即保留该路径）
            return !(path.toString().matches(reg));
        } catch (IOException e) {
            e.printStackTrace();
            return false; // 出现异常时，跳过该路径
        }
    }
}

/**
 * 利用 FSDataOutputStream 和 FSDataInputStream 合并 HDFS 中的文件。
 */
public class MergeFile {
    Path inputPath = null; // 待合并的文件所在的目录的路径
    Path outputPath = null; // 输出文件的路径

    /**
     * 构造函数，初始化输入和输出路径。
     * @param input 输入目录的路径
     * @param output 输出文件的路径
     */
    public MergeFile(String input, String output) {
        this.inputPath = new Path(input);
        this.outputPath = new Path(output);
    }

    /**
     * 执行文件合并操作。
     * @throws IOException 如果发生 I/O 错误
     */
    public void doMerge() throws IOException {
        Configuration conf = new Configuration(); // 创建 Hadoop 配置对象

        // 设置 HDFS 的默认文件系统 URI
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");

        // 获取输入路径和输出路径的文件系统对象
        FileSystem fsSource = FileSystem.get(URI.create(inputPath.toString()), conf);
        FileSystem fsDst = FileSystem.get(URI.create(outputPath.toString()), conf);

        // 使用自定义过滤器过滤掉后缀为 .abc 的文件，并跳过目录
        FileStatus[] sourceStatus = fsSource.listStatus(inputPath,
                new MyPathFilter(".*\\.abc")); // 过滤规则：排除后缀为 .abc 的文件

        // 创建输出文件的写入流
        FSDataOutputStream fsdos = fsDst.create(outputPath);

        // 创建打印流，将内容同时输出到控制台
        PrintStream ps = new PrintStream(System.out);

        // 遍历过滤后的文件状态数组
        for (FileStatus sta : sourceStatus) {
            // 打印文件路径、大小和权限信息
            System.out.print("路径：" + sta.getPath() + "    文件大小：" + sta.getLen()
                    + "   权限：" + sta.getPermission() + "   内容：");

            // 打开当前文件的输入流
            FSDataInputStream fsdis = fsSource.open(sta.getPath());

            byte[] data = new byte[1024]; // 缓冲区，每次读取 1024 字节
            int read = -1;

            // 循环读取文件内容
            while ((read = fsdis.read(data)) > 0) {
                ps.write(data, 0, read); // 将内容写入控制台
                fsdos.write(data, 0, read); // 将内容写入输出文件
            }

            fsdis.close(); // 关闭当前文件的输入流
        }

        ps.close(); // 关闭打印流
        fsdos.close(); // 关闭输出文件的写入流
    }

    /**
     * 主方法，程序入口。
     * @param args 命令行参数
     * @throws IOException 如果发生 I/O 错误
     */
    public static void main(String[] args) throws IOException {
        // 初始化 MergeFile 对象，指定输入目录和输出文件路径
        MergeFile merge = new MergeFile(
                "hdfs://localhost:9000/user/hadoop/", // 输入目录
                "hdfs://localhost:9000/user/hadoop/merge.txt"); // 输出文件

        // 执行文件合并操作
        merge.doMerge();
    }
}
